package cn.chen.sparkproject.spark;

import cn.chen.sparkproject.conf.ConfigurationManager;
import cn.chen.sparkproject.constant.Constants;
import cn.chen.sparkproject.dao.*;
import cn.chen.sparkproject.domain.*;
import cn.chen.sparkproject.factory.DAOFactory;
import cn.chen.sparkproject.test.MockData;
import cn.chen.sparkproject.util.*;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.spark.Accumulator;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.hive.HiveContext;
import scala.Tuple2;
import com.google.common.base.Optional;

import java.util.*;

/**
 * 用户访问session分析spark作业
 *
 * 1.时间范围
 * 2.性别
 * 3.年龄范围
 * 4.职业：多选
 * 5.城市：多选
 * 6.搜索词
 * 7.点击品类
 */
public class UserVisitSessionAnalyzeSpark {

    public static void main(String[] args) {

        args = new String[]{"2"};

        // 构建spark上下文
        SparkConf conf = new SparkConf()
                .setAppName(Constants.SPARK_APP_SESSION_NAME)
                .setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = getSQLContext(sc.sc());

        // 生成模拟数据
        mockData(sc, sqlContext);
        // 获取DAO
        ITaskDAO taskDAO = DAOFactory.getTaskDao();

        // 查询指定任务
        long taskId = ParamUtils.getTaskIdFromArgs(args);
        Task task = taskDAO.findById(taskId);
        JSONObject taskParam = JSON.parseObject(task.getTaskParam());

        // session粒度数据聚合，从user_visit_action表中查询指定日期的行为数据
        JavaRDD<Row> actionRDD = getActionRDDByDateRange(sqlContext, taskParam);
        JavaPairRDD<String, Row> sessionid2actionRDD = getSessionid2actionRDD(actionRDD);
        // 将行为数据按照session_id进行groupBy分组
        // 与用户数据join
        // 获得session粒度信息，同时包含用户数据
        // return <sessionid,fullAggrInfo>
        JavaPairRDD<String, String> sessionid2AggrInfoRDD = aggregateBySession(sqlContext, actionRDD);

        for(Tuple2<String, String> tuple : sessionid2AggrInfoRDD.take(10)){
            System.out.println(tuple._2);
        }

        // 累加器
        Accumulator<String> sessionAggrStatAccumulator = sc.accumulator(
                "", new SessionAggrStatAccumulator());
        // 过滤
        JavaPairRDD<String, String> filteredSessionid2AggrInfoRDD =
                filteredSessionAndAggrStat(sessionid2AggrInfoRDD, taskParam,
                        sessionAggrStatAccumulator);

        // 通过筛选条件的访问明细数据
        JavaPairRDD<String, Row> sessionid2detailRDD = getSessionid2detailRDD(filteredSessionid2AggrInfoRDD, sessionid2actionRDD);

        System.out.println(filteredSessionid2AggrInfoRDD.count());

        randomExtractSession(task.getTaskId(),filteredSessionid2AggrInfoRDD,sessionid2actionRDD);

        // 计算出各个范围session的占比并写入MySQL
        calculateAndPersistAggrStat(sessionAggrStatAccumulator.value(), taskId);

        // 获取top10热门品类
        List<Tuple2<CategorySortKey, String>> top10CategoryList = getTop10Category(sessionid2detailRDD, task.getTaskId());

        // 获取top10活跃session
        getTop10Session(sc, top10CategoryList, sessionid2detailRDD ,task.getTaskId());

        // 关闭Spark上下文
        sc.close();
    }

    /**
     * 获取SqlContext
     * 本地测试生成sqlContext对象，生产环境运行生成HiveContext对象
     * @param sc
     * @return
     */
    private static SQLContext getSQLContext(SparkContext sc){
        boolean local = ConfigurationManager.getBoolean(Constants.SPARK_LOCAL);
        if(local){
            return new SQLContext(sc);
        }else{
            return new HiveContext(sc);
        }
    }

    /**
     * 生成模拟数据（本地模式）
     * @param sc
     * @param sqlContext
     */
    private static void mockData(JavaSparkContext sc, SQLContext sqlContext){
        boolean local = ConfigurationManager.getBoolean(Constants.SPARK_LOCAL);
        if(local){
            MockData.mock(sc, sqlContext);
        }
    }

    /**
     * 获取指定日期范围内的用户访问行为数据
     * @param sqlContext
     * @param taskParam
     * @return
     */
    private static JavaRDD<Row> getActionRDDByDateRange(
            SQLContext sqlContext, JSONObject taskParam){
        String startDate = ParamUtils.getParam(taskParam, Constants.PARAM_START_DATE);
        String endDate = ParamUtils.getParam(taskParam, Constants.PARAM_END_DATE);
        String sql = "select * " +
                "from user_visit_action " +
                "where date >= '" + startDate +"' " +
                "and date <= '" + endDate +"'";
        DataFrame actionDF = sqlContext.sql(sql);
        return actionDF.javaRDD();
    }


    private static JavaPairRDD<String, Row> getSessionid2actionRDD(JavaRDD<Row> actionRDD) {
        return actionRDD.mapToPair(
            new PairFunction<Row, String, Row>() {
                @Override
                public Tuple2<String, Row> call(Row row) throws Exception {
                    return new Tuple2<String, Row>(row.getString(2), row);
                }
        });
    }

    /**
     * session粒度聚合
     * @param actionRDD
     * @return
     */
    private static JavaPairRDD<String, String> aggregateBySession(
            SQLContext sqlContext, final JavaRDD<Row> actionRDD){
        // <sessionid, row>
        JavaPairRDD<String, Row> sessionid2ActionRDD = actionRDD.mapToPair(
                new PairFunction<Row, String, Row>() {
                    private static final long serialVersionUID = -3416936267785977789L;

                    @Override
            public Tuple2<String, Row> call(Row row) throws Exception {
                return new Tuple2<String, Row>(row.getString(2), row);
            }
        });

        // session粒度分组<<sessionid>,Iterable<Row>>
        JavaPairRDD<String, Iterable<Row>> sessionid2ActionsRDD = sessionid2ActionRDD.groupByKey();

        // 对分组聚合<userid,partAggrInfo(sessionid, searchKeyWords, clickCategoryIds)>
        JavaPairRDD<Long, String> userid2PartAggrInfoRDD = sessionid2ActionsRDD.mapToPair(
            new PairFunction<Tuple2<String,Iterable<Row>>, Long, String>() {
                private static final long serialVersionUID = 959391019630160709L;

                    @Override
                public Tuple2<Long, String> call(Tuple2<String, Iterable<Row>> tuple) throws Exception {
                    String sessionid = tuple._1;
                    Iterator<Row> iterator = tuple._2.iterator();

                    Long userid = null;
                    StringBuffer searchKeywordBuffer = new StringBuffer("");
                    StringBuffer clickCategoryIdsBuffer = new StringBuffer("");
                    Date startTime = null;
                    Date endTime = null;
                    int stepLength = 0;

                    while(iterator.hasNext()){
                        Row row = iterator.next();
                        if(userid == null){
                            userid = row.getLong(1);
                        }
                        String searchKeyword  = row.getString(5);
                        Long clickCategoryId = row.getLong(6);

                        if(StringUtils.isNotEmpty(searchKeyword)){
                            if(!searchKeywordBuffer.toString().contains(searchKeyword)){
                                searchKeywordBuffer.append(searchKeyword + ",");
                            }
                        }

                        if(clickCategoryId != null){
                            if(!clickCategoryIdsBuffer.toString().contains(String.valueOf(
                                    clickCategoryId))){
                                clickCategoryIdsBuffer.append(clickCategoryId + ",");
                            }
                        }

                        Date actionTime = DateUtils.parseTime(row.getString(4));

                        if(startTime == null){
                            startTime = actionTime;
                        }

                        if(endTime == null){
                            endTime = actionTime;
                        }

                        if(actionTime.before(startTime)){
                            startTime = actionTime;
                        }

                        if(actionTime.after(endTime)){
                            endTime = actionTime;
                        }

                        stepLength ++;

                    }

                    String searchKeywords = StringUtils.trimComma(searchKeywordBuffer.toString());
                    String clickCategoryIds = StringUtils.trimComma(clickCategoryIdsBuffer.toString());

                    long visitLength = (endTime.getTime() - startTime.getTime()) / 1000;

                    String partAggrInfo = Constants.FIELD_SESSION_ID + "=" + sessionid + "|" +
                            Constants.FIELD_SEARCH_KEYWORDS + "=" + searchKeywords + "|" +
                            Constants.FIELD_CLICK_CATEGORY_IDS + "=" + clickCategoryIds + "|" +
                            Constants.FIELD_VISIT_LENGTH + "=" + visitLength + "|" +
                            Constants.FIELD_STEP_LENGTH + "=" + stepLength + "|" +
                            Constants.FIELD_START_TIME + "=" + DateUtils.formatTime(startTime);
                    // <userid,partAggrInfo(sessionid, searchKeyWords, clickCategoryIds)>
                    return new Tuple2<Long, String>(userid, partAggrInfo);
                }
            });

        // 查询所有用户数据<userid, row>
        String sql = "select * from user_info";
        final JavaRDD<Row> userInfoRDD = sqlContext.sql(sql).javaRDD();
        JavaPairRDD<Long, Row> userid2InfoRDD = userInfoRDD.mapToPair(
                new PairFunction<Row, Long, Row>() {
                    private static final long serialVersionUID = 1268552843758806362L;

                    @Override
            public Tuple2<Long, Row> call(Row row) throws Exception {
                return new Tuple2<Long, Row>(row.getLong(0), row);
            }
        });

        // session粒度聚合数据和用户数据join
        JavaPairRDD<Long, Tuple2<String, Row>> userid2FullInfoRDD =
                userid2PartAggrInfoRDD.join(userid2InfoRDD);
        // 对join起来的数据进行拼接并且返回<sessionid,fullAggrInfo>
        JavaPairRDD<String, String> sessionid2FullAggrInfoRDD = userid2FullInfoRDD.mapToPair(
            new PairFunction<Tuple2<Long, Tuple2<String, Row>>, String, String>() {
                private static final long serialVersionUID = -8349428342341439203L;

                @Override
                public Tuple2<String, String> call(Tuple2<Long, Tuple2<String, Row>> tuple)
                        throws Exception {
                    String partAggrInfo = tuple._2._1;
                    Row userInfoRow = tuple._2._2;

                    String sessionid = StringUtils.getFieldFromConcatString(
                            partAggrInfo, "\\|", Constants.FIELD_SESSION_ID);

                    int age = userInfoRow.getInt(3);
                    String professional = userInfoRow.getString(4);
                    String city = userInfoRow.getString(5);
                    String sex = userInfoRow.getString(6);

                    String fullAggrInfo = partAggrInfo + "|"
                            + Constants.FIELD_AGE + "=" + age + "|"
                            + Constants.FIELD_PROFESSIONAL + "=" + professional + "|"
                            + Constants.FIElD_CITY + "=" + city + "|"
                            + Constants.FIELD_SEX + "=" + sex;

                    return new Tuple2<String, String>(sessionid, fullAggrInfo);
                }
    });
        return sessionid2FullAggrInfoRDD;
    }

    private static JavaPairRDD<String, String> filteredSessionAndAggrStat(
            JavaPairRDD<String, String> sessionid2AggrInfoRDD,
            final JSONObject taskParam, final Accumulator<String> sessionAggrStatAccumulator){

        String startAge = ParamUtils.getParam(taskParam, Constants.PARAM_START_AGE);
        String endAge = ParamUtils.getParam(taskParam, Constants.PARAM_END_AEG);
        String profesisonals = ParamUtils.getParam(taskParam, Constants.PARAM_PROFESSIONALS);
        String cities = ParamUtils.getParam(taskParam, Constants.PARAM_CITY);
        String sex = ParamUtils.getParam(taskParam, Constants.PARAM_SEX);
        String keywords = ParamUtils.getParam(taskParam, Constants.PARAM_KEYWORDS);
        String categoryIds = ParamUtils.getParam(taskParam, Constants.PARAM_CATEGORY_IDS);
        String _paramer = (startAge != null ? Constants.PARAM_START_AGE + "=" + startAge + "|" : "")
                + (endAge != null ? Constants.PARAM_END_AEG + "=" + endAge + "|" : "")
                + (profesisonals != null ? Constants.PARAM_PROFESSIONALS + "=" + profesisonals + "|" : "")
                + (cities != null ? Constants.PARAM_CITY + "=" + cities + "|" : "")
                + (sex != null ? Constants.PARAM_SEX + "=" + sex + "|" : "")
                + (keywords != null ? Constants.PARAM_KEYWORDS + "=" + keywords + "|" : "")
                + (categoryIds != null ? Constants.PARAM_CATEGORY_IDS + "=" + categoryIds : "");

        if(_paramer.endsWith("\\|")){
            _paramer.substring(0, _paramer.length() - 1);
        }

        final String paramer = _paramer;

        JavaPairRDD<String, String> filteredSession2AggrInfoRDD = sessionid2AggrInfoRDD.filter(
            new Function<Tuple2<String, String>, Boolean>() {
                private static final long serialVersionUID = -8850224345609975472L;

                @Override
                public Boolean call(Tuple2<String, String> tuple) throws Exception {
                    String aggrInfo = tuple._2;

                    // 过滤
                    int age = Integer.valueOf(StringUtils.getFieldFromConcatString(
                            aggrInfo, "\\|", Constants.FIELD_AGE));
                    String startAge = ParamUtils.getParam(taskParam, Constants.PARAM_START_AGE);
                    String endAge = ParamUtils.getParam(taskParam, Constants.PARAM_END_AEG);

                    if(!ValidUtils.between(aggrInfo, Constants.FIELD_AGE
                            ,paramer, Constants.PARAM_START_AGE,Constants.PARAM_END_AEG)){
                        return false;
                    }

                    if(!ValidUtils.in(aggrInfo, Constants.FIELD_PROFESSIONAL
                            ,paramer, Constants.PARAM_PROFESSIONALS)){
                        return false;
                    }

                    if(!ValidUtils.in(aggrInfo, Constants.FIElD_CITY
                            , paramer, Constants.PARAM_SEX)){
                        return false;
                    }

                    if(!ValidUtils.equal(aggrInfo, Constants.FIELD_SEX
                            ,paramer,Constants.PARAM_SEX)){
                        return false;
                    }

                    if(!ValidUtils.in(aggrInfo, Constants.FIELD_SEARCH_KEYWORDS
                            ,paramer,Constants.PARAM_KEYWORDS)){
                        return false;
                    }

                    if(!ValidUtils.in(aggrInfo, Constants.FIELD_CLICK_CATEGORY_IDS
                        ,paramer,Constants.PARAM_CATEGORY_IDS)){
                        return false;
                    }
        //                if(startAge != null && endAge != null){
        //                    if(age >= Integer.parseInt(startAge) && age <= Integer.parseInt(endAge)){
        //
        //                    }
        //                }

                    // 对过滤后的session进行累加计数
                    sessionAggrStatAccumulator.add(Constants.SESSION_COUNT);

                    long visitLength = Long.valueOf(StringUtils.getFieldFromConcatString(aggrInfo
                            , "\\|", Constants.FIELD_VISIT_LENGTH));
                    long stepLength = Long.valueOf(StringUtils.getFieldFromConcatString(aggrInfo
                            , "\\|", Constants.FIELD_STEP_LENGTH));

                    calculateVisitLength(visitLength);
                    calculateStepLength(stepLength);

                    return true;
                }

            private void calculateVisitLength(long visitLength){
                if(visitLength >= 1 && visitLength <= 3){
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_1s_3s);
                } else if(visitLength >= 4 && visitLength <= 6){
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_4s_6s);
                } else if(visitLength >= 7 && visitLength <= 9){
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_7s_9s);
                } else if(visitLength >= 10 && visitLength <= 30){
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_10s_30s);
                } else if(visitLength > 30 && visitLength <= 60){
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30s_60s);
                } else if(visitLength > 60 && visitLength <= 180){
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_1m_3m);
                } else if(visitLength > 180 && visitLength <= 600){
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_3m_10m);
                } else if(visitLength > 600 && visitLength <= 1800){
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_10m_30m);
                } else if(visitLength > 1800){
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30m);
                }
            }

            private void calculateStepLength(long stepLength){
                if(stepLength >= 1 && stepLength <= 3){
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_1_3);
                } else if(stepLength >= 4 && stepLength <= 6){
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_4_6);
                } else if(stepLength >= 7 && stepLength <= 9){
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_7_9);
                } else if(stepLength >= 10 && stepLength <= 30){
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_10_30);
                } else if(stepLength > 30 && stepLength <= 60){
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_30_60);
                } else if(stepLength > 60){
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_60);
                }
            }

        });
        return filteredSession2AggrInfoRDD;
    }

    /**
     * 随机抽取session
     * @param sessionid2AggrInfoRDD
     * @param sessionid2actionRDD
     */
    private static void randomExtractSession(final long taskid, JavaPairRDD<String, String> sessionid2AggrInfoRDD, JavaPairRDD<String, Row> sessionid2actionRDD) {
        // 计算每天每小时的session数量<yyyy-MM-dd, sessionid>
        JavaPairRDD<String, String> time2sessionidRDD = sessionid2AggrInfoRDD.mapToPair(
            new PairFunction<Tuple2<String, String>, String, String>() {
                private static final long serialVersionUID = -4122016613725307586L;

                @Override
                public Tuple2<String, String> call(Tuple2<String, String> tuple) throws Exception {
                    String aggrInfo = tuple._2;
                    String startTime = StringUtils.getFieldFromConcatString(
                            aggrInfo, "\\|", Constants.FIELD_START_TIME);
                    String dateHour = DateUtils.getDateHour(startTime);
                    return new Tuple2<String, String>(dateHour, aggrInfo);
                }
        });

        Map<String, Object> countMap = time2sessionidRDD.countByKey();

        // 使用按时间比例随机抽取算法，计算每天每小时要抽取session的索引
        // <yyyy-MM-dd, count> -> <yyyy-MM-dd, <HH, count>>
        Map<String, Map<String, Long>> dateHourCountMap =
                new HashMap<String, Map<String, Long>>();
        for(Map.Entry<String, Object> countEntry : countMap.entrySet()){
            String dateHour = countEntry.getKey();
            String date = dateHour.split("_")[0];
            String hour = dateHour.split("_")[1];

            long count = Long.valueOf(String.valueOf(countEntry.getValue()));

            Map<String, Long> hourCountMap = dateHourCountMap.get(date);
            if(hourCountMap == null){
                hourCountMap = new HashMap<String, Long>();
                dateHourCountMap.put(date, hourCountMap);
            }

            hourCountMap.put(hour, count);

        }

        int extractNumberPerDay = 100 / dateHourCountMap.size();

        // <date, <hour,(3,5,20,102)>>
        final Map<String, Map<String, List<Integer>>> dateHourExtractMap =
                new HashMap<String, Map<String, List<Integer>>>();

        Random random = new Random();

        for(Map.Entry<String, Map<String, Long>> dateHourCountEntry : dateHourCountMap.entrySet()){
            String date = dateHourCountEntry.getKey();
            Map<String, Long> hourCountMap = dateHourCountEntry.getValue();

            // 计算出一天中的session总数
            long sessionCount = 0L;
            for(long hourCount : hourCountMap.values()){
                sessionCount += hourCount;
            }

            Map<String, List<Integer>> hourExtractMap = dateHourExtractMap.get(date);
            if(hourExtractMap == null){
                hourExtractMap = new HashMap<String, List<Integer>>();
                dateHourExtractMap.put(date, hourExtractMap);
            }

            // 遍历每个小时
            for(Map.Entry<String,Long> hourCountEntry : hourCountMap.entrySet()){
                String hour = hourCountEntry.getKey();
                long count = hourCountEntry.getValue();

                int hourExtractNumber = (int) (((double)count / (double)sessionCount) * extractNumberPerDay);

                 List<Integer> extractIndexList = hourExtractMap.get(hour);
                 if(extractIndexList == null){
                     extractIndexList = new ArrayList<Integer>();
                     hourExtractMap.put(hour, extractIndexList);
                 }

                if(hourExtractNumber > count){
                    hourExtractNumber = (int)count;
                }

                for(int i = 0; i < hourExtractNumber; i ++){
                    int extractIndex = random.nextInt((int) count);
                    while(extractIndexList.contains(extractIndex)){
                        extractIndex = random.nextInt((int) count);
                    }
                    extractIndexList.add(extractIndex);
                }
            }
        }

        // <dateHour, (session aggrInfo)>
        JavaPairRDD<String, Iterable<String>> time2sessionsRDD = time2sessionidRDD.groupByKey();
        JavaPairRDD<String, String> extractSessionidsRDD = time2sessionsRDD.flatMapToPair(
            new PairFlatMapFunction<Tuple2<String,Iterable<String>>, String, String>() {
                private static final long serialVersionUID = 8850393751207239987L;

                @Override
                public Iterable<Tuple2<String, String>> call(Tuple2<String, Iterable<String>> tuple) throws Exception {
                        List<Tuple2<String, String>> extractSessionids =
                                new ArrayList<Tuple2<String, String>>();
                        String dateHour = tuple._1;
                        String date = dateHour.split("_")[0];
                        String hour = dateHour.split("_")[1];

                        Iterator<String> iterator = tuple._2.iterator();
                        List<Integer> extractIndexList = dateHourExtractMap.get(date).get(hour);

                        ISessionRandomExtractDao sessionRandomExtractDao =
                                DAOFactory.getSessionRandomExtractDao();

                        int index = 0;
                        while(iterator.hasNext()){
                            String sessionAggrInfo = iterator.next();
                            if(extractIndexList.contains(index)){
                                String sessionid = StringUtils.getFieldFromConcatString(
                                        sessionAggrInfo, "\\|", Constants.FIELD_SESSION_ID);
                                SessionRandomExtract sessionRandomExtract = new SessionRandomExtract();
                                sessionRandomExtract.setTaskid(taskid);
                                sessionRandomExtract.setSessionid(sessionid);
                                sessionRandomExtract.setStartTime(StringUtils.getFieldFromConcatString(
                                        sessionAggrInfo, "\\|", Constants.FIELD_START_TIME));
                                sessionRandomExtract.setSearchKeyWords(StringUtils.getFieldFromConcatString(
                                        sessionAggrInfo, "\\|", Constants.FIELD_SEARCH_KEYWORDS));
                                sessionRandomExtract.setClickCategoryIds(StringUtils.getFieldFromConcatString(
                                        sessionAggrInfo,  "\\|", Constants.FIELD_CLICK_CATEGORY_IDS));
                                sessionRandomExtractDao.insert(sessionRandomExtract);
                                extractSessionids.add(new Tuple2<String, String>(sessionid, sessionid));
                            }
                            index ++;
                        }
                        return extractSessionids;
                }
            });

        // 明细数据
        JavaPairRDD<String, Tuple2<String, Row>> extractSessionDetailRDD =
                extractSessionidsRDD.join(sessionid2actionRDD);

        extractSessionDetailRDD.foreach(
            new VoidFunction<Tuple2<String, Tuple2<String, Row>>>() {
                private static final long serialVersionUID = 2041602725208800326L;

                @Override
                public void call(Tuple2<String, Tuple2<String, Row>> tuple) throws Exception {
                    Row row = tuple._2._2;

                    SessionDetail sessionDetail = new SessionDetail();
                    sessionDetail.setTaskid(taskid);
                    sessionDetail.setUserid(row.getLong(1));
                    sessionDetail.setSessionid(row.getString(2));
                    sessionDetail.setPageid(row.getLong(3));
                    sessionDetail.setActionTime(row.getString(4));
                    sessionDetail.setSearchKeyWord(row.getString(5));
                    sessionDetail.setClickCategoryId(row.getLong(6));
                    sessionDetail.setClickProductId(row.getLong(7));
                    sessionDetail.setOrderCategoryIds(row.getString(8));
                    sessionDetail.setOrderProductIds(row.getString(9));
                    sessionDetail.setPayCategoryIds(row.getString(10));
                    sessionDetail.setPayProductIds(row.getString(11));

                    ISessionDetailDao sessionDetailDao = DAOFactory.getSessionDetailDao();
                    sessionDetailDao.insert(sessionDetail);
                }
        });

    }

    /**
     * 计算各个session占比并写入MySQL
     * @param value
     */
    private static void calculateAndPersistAggrStat(String value, long taskid) {
        long session_count = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.SESSION_COUNT));
        long visit_length_1s_3s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_1s_3s));
        long visit_length_4s_6s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_4s_6s));
        long visit_length_7s_9s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_7s_9s));
        long visit_length_10s_30s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_10s_30s));
        long visit_length_30s_60s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_30s_60s));
        long visit_length_1m_3m = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_1m_3m));
        long visit_length_3m_10m = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_3m_10m));
        long visit_length_10m_30m = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_10m_30m));
        long visit_length_30m = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_30m));

        long step_length_1_3 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_1_3));
        long step_length_4_6 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_4_6));
        long step_length_7_9 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_7_9));
        long step_length_10_30 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_10_30));
        long step_length_30_60 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_30_60));
        long step_length_60 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_60));

        double visit_length_1s_3s_ratio = NumberUtils.formatDouble(
                (double)visit_length_1s_3s / (double)session_count, 2);
        double visit_length_4s_6s_ratio = NumberUtils.formatDouble(
                (double)visit_length_4s_6s/ (double)session_count, 2);
        double visit_length_7s_9s_ratio = NumberUtils.formatDouble(
                (double)visit_length_7s_9s / (double)session_count, 2);
        double visit_length_10s_30s_ratio = NumberUtils.formatDouble(
                (double)visit_length_10s_30s / (double)session_count, 2);
        double visit_length_30s_60s_ratio = NumberUtils.formatDouble(
                (double)visit_length_30s_60s / (double)session_count, 2);
        double visit_length_1m_3m_ratio = NumberUtils.formatDouble(
                (double)visit_length_1m_3m / (double)session_count, 2);
        double visit_length_3m_10m_ratio = NumberUtils.formatDouble(
                (double)visit_length_3m_10m / (double)session_count, 2);
        double visit_length_10m_30m_ratio = NumberUtils.formatDouble(
                (double)visit_length_10m_30m / (double)session_count, 2);
        double visit_length_30m_ratio = NumberUtils.formatDouble(
                (double)visit_length_30m / (double)session_count, 2);
        double step_length_1_3_ratio = NumberUtils.formatDouble(
                (double)step_length_1_3 / (double)session_count, 2);
        double step_length_4_6_ratio = NumberUtils.formatDouble(
                (double)step_length_4_6 / (double)session_count, 2);
        double step_length_7_9_ratio = NumberUtils.formatDouble(
                (double)step_length_7_9 / (double)session_count, 2);
        double step_length_10_30_ratio = NumberUtils.formatDouble(
                (double)step_length_10_30 / (double)session_count, 2);
        double step_length_30_60_ratio = NumberUtils.formatDouble(
                (double)step_length_30_60 / (double)session_count, 2);
        double step_length_60_ratio = NumberUtils.formatDouble(
                (double)step_length_60 / (double)session_count, 2);


        SessionAggrStat sessionAggrStat = new SessionAggrStat();
        sessionAggrStat.setTaskid(taskid);
        sessionAggrStat.setSessionCount(session_count);
        sessionAggrStat.setVisit_length_1s_3s_radio(visit_length_1s_3s_ratio);
        sessionAggrStat.setVisit_length_4s_6s_radio(visit_length_4s_6s_ratio);
        sessionAggrStat.setVisit_length_7s_9s_radio(visit_length_7s_9s_ratio);
        sessionAggrStat.setVisit_length_10s_30s_radio(visit_length_10s_30s_ratio);
        sessionAggrStat.setVisit_length_30s_60s_radio(visit_length_30s_60s_ratio);
        sessionAggrStat.setVisit_length_1m_3m_radio(visit_length_1m_3m_ratio);
        sessionAggrStat.setVisit_length_3m_10m_radio(visit_length_3m_10m_ratio);
        sessionAggrStat.setVisit_length_10m_30m_radio(visit_length_10m_30m_ratio);
        sessionAggrStat.setVisit_length_30m_radio(visit_length_30m_ratio);
        sessionAggrStat.setStep_length_1_3_radio(step_length_1_3_ratio);
        sessionAggrStat.setStep_length_4_6_radio(step_length_4_6_ratio);
        sessionAggrStat.setStep_length_7_9_radio(step_length_7_9_ratio);
        sessionAggrStat.setStep_length_10_30_radio(step_length_10_30_ratio);
        sessionAggrStat.setStep_length_30_60_radio(step_length_30_60_ratio);
        sessionAggrStat.setStep_length_60_radio(step_length_60_ratio);

        ISessionAggrStatDao sessionAggrStatDao = DAOFactory.getSessionAggrStatDao();
        sessionAggrStatDao.insert(sessionAggrStat);
    }

    private static JavaPairRDD<String, Row> getSessionid2detailRDD(JavaPairRDD<String, String> filteredSessionid2AggrInfoRDD,
                                                                   JavaPairRDD<String, Row> sessionid2actionRDD){
        JavaPairRDD<String, Row> sessionid2detailRDD = filteredSessionid2AggrInfoRDD.join(sessionid2actionRDD)
                .mapToPair(
                        new PairFunction<Tuple2<String, Tuple2<String, Row>>, String, Row>() {
                            private static final long serialVersionUID = -4195505523511186534L;

                            @Override
                            public Tuple2<String, Row> call(Tuple2<String, Tuple2<String, Row>> tuple) throws Exception {
                                return new Tuple2<String, Row>(tuple._1, tuple._2._2);
                            }
                        });
        return sessionid2detailRDD;
    }

    private static List<Tuple2<CategorySortKey, String>> getTop10Category(
            JavaPairRDD<String, Row> sessionid2detailRDD, long taskId) {

        // 获取session访问过(点击，下单，支付)的所有品类id
        JavaPairRDD<Long, Long> categoryidRDD = sessionid2detailRDD.flatMapToPair(
            new PairFlatMapFunction<Tuple2<String, Row>, Long, Long>() {
                private static final long serialVersionUID = 1438816983763272368L;

                @Override
                public Iterable<Tuple2<Long, Long>> call(Tuple2<String, Row> tuple) throws Exception {
                    Row row = tuple._2;
                    List<Tuple2<Long, Long>> list = new ArrayList<Tuple2<Long, Long>>();

                    Long clickCategoryId = row.getLong(6);
                    if(clickCategoryId != null){
                        list.add(new Tuple2<Long, Long>(clickCategoryId, clickCategoryId));
                    }

                    String orderCategory = row.getString(8);
                    if(orderCategory != null){
                        String[] orderCategoryIdsSplited = orderCategory.split(",");
                        for(String orderCategoryId : orderCategoryIdsSplited){
                            list.add(new Tuple2<Long, Long>(
                                    Long.valueOf(orderCategoryId), Long.valueOf(orderCategoryId)));
                        }
                    }

                    String payCategoryIds = row.getString(10);
                    if(payCategoryIds != null){
                        String[] payCategoryIdsSplited = payCategoryIds.split(",");
                        for(String paycategoryId : payCategoryIdsSplited){
                            list.add(new Tuple2<Long, Long>(
                                    Long.valueOf(paycategoryId), Long.valueOf(paycategoryId)));
                        }
                    }
                    return list;
                }
        });

        categoryidRDD = categoryidRDD.distinct();

        JavaPairRDD<Long, Long> clickCategoryId2CountRDD =
                getClickCategoryId2CountRDD(sessionid2detailRDD);
        JavaPairRDD<Long, Long> orderCategoryId2CountRDD =
                getOrderCategoryId2CountRDD(sessionid2detailRDD);
        JavaPairRDD<Long, Long> payCategoryId2CountRDD =
                getPayCategoryId2CountRDD(sessionid2detailRDD);

        JavaPairRDD<Long, String> categoryid2countRDD = joinCategoryAndData(categoryidRDD, clickCategoryId2CountRDD,
                orderCategoryId2CountRDD, payCategoryId2CountRDD);

        JavaPairRDD<CategorySortKey, String> sortkey2countRDD = categoryid2countRDD.mapToPair(
            new PairFunction<Tuple2<Long, String>, CategorySortKey, String>() {
                private static final long serialVersionUID = -3292136461246399894L;

                @Override
                public Tuple2<CategorySortKey, String> call(Tuple2<Long, String> tuple) throws Exception {
                    String countInfo  = tuple._2;
                    long clickCount = Long.valueOf(StringUtils.getFieldFromConcatString(
                            countInfo, "\\|", Constants.FIELD_CLICK_COUNT));
                    long orderCount = Long.valueOf(StringUtils.getFieldFromConcatString(
                            countInfo, "\\|", Constants.FIELD_ORDER_COUNT));
                    long payCount = Long.valueOf(StringUtils.getFieldFromConcatString(
                            countInfo, "\\|", Constants.FIELD_PAY_COUNT));

                    CategorySortKey categorySortKey = new CategorySortKey(clickCount, orderCount, payCount);

                    return new Tuple2<CategorySortKey, String>(categorySortKey, countInfo);
                }
        });

        JavaPairRDD<CategorySortKey, String> sortedCategoryCountRDD= sortkey2countRDD.sortByKey(false);
        List<Tuple2<CategorySortKey, String>> top10CategoryList = sortedCategoryCountRDD.take(10);
        ITop10CategoryDao categoryDao = DAOFactory.getTop10CategoryDao();

        for(Tuple2<CategorySortKey, String> tuple : top10CategoryList){
            String countInfo = tuple._2;
            long categoryid = Long.valueOf(StringUtils.getFieldFromConcatString(countInfo, "\\|", Constants.FIELD_CATEGORY_ID));
            long clickCount = Long.valueOf(StringUtils.getFieldFromConcatString(
                    countInfo, "\\|", Constants.FIELD_CLICK_COUNT));
            long orderCount = Long.valueOf(StringUtils.getFieldFromConcatString(
                    countInfo, "\\|", Constants.FIELD_ORDER_COUNT));
            long payCount = Long.valueOf(StringUtils.getFieldFromConcatString(
                    countInfo, "\\|", Constants.FIELD_PAY_COUNT));

            Top10Category category = new Top10Category();
            category.setTaskid(taskId);
            category.setCategoryid(categoryid);
            category.setClickCount(clickCount);
            category.setOrderCount(orderCount);
            category.setPayCount(payCount);
            categoryDao.insert(category);
        }
        return top10CategoryList;

    }

    private static JavaPairRDD<Long, Long> getClickCategoryId2CountRDD(JavaPairRDD<String, Row> sessionid2detailRDD){
        JavaPairRDD<String, Row> clickActionRDD = sessionid2detailRDD.filter(
            new Function<Tuple2<String, Row>, Boolean>() {
                private static final long serialVersionUID = -1458119591404592760L;

                @Override
                public Boolean call(Tuple2<String, Row> tuple) throws Exception {
                    Row row = tuple._2;
                    return row.get(6) != null ? true : false;
                }
            });

        JavaPairRDD<Long, Long> clickCategoryIdRDD = clickActionRDD.mapToPair(
            new PairFunction<Tuple2<String,Row>, Long, Long>() {
                private static final long serialVersionUID = 9080259648852803790L;

                @Override
                public Tuple2<Long, Long> call(Tuple2<String, Row> tuple) throws Exception {
                    long clickCategoryId = tuple._2.getLong(6);
                    return new Tuple2<Long, Long>(clickCategoryId, 1L);
                }
            });

        JavaPairRDD<Long, Long> clickCategoryId2CountRDD = clickCategoryIdRDD.reduceByKey(
            new Function2<Long, Long, Long>() {
                private static final long serialVersionUID = 195209443714231070L;

                @Override
                public Long call(Long v1, Long v2) throws Exception {
                    return v1 + v2;
                }
            });

        return clickCategoryId2CountRDD;
    }

    private static JavaPairRDD<Long, Long> getOrderCategoryId2CountRDD(JavaPairRDD<String, Row> sessionid2detailRDD) {
        JavaPairRDD<String, Row> orderActionRDD = sessionid2detailRDD.filter(
            new Function<Tuple2<String, Row>, Boolean>() {
                private static final long serialVersionUID = -1458119591404592760L;

                @Override
                public Boolean call(Tuple2<String, Row> tuple) throws Exception {
                    Row row = tuple._2;
                    return row.getString(8) != null ? true : false;
                }
            });

        JavaPairRDD<Long, Long> orderCategoryIdRDD = orderActionRDD.flatMapToPair(
            new PairFlatMapFunction<Tuple2<String, Row>, Long, Long>() {
                private static final long serialVersionUID = -6783683443025609669L;

                @Override
                public Iterable<Tuple2<Long, Long>> call(Tuple2<String, Row> tuple) throws Exception {
                    Row row = tuple._2;
                    String orderCategoryIds = row.getString(8);
                    String[] orderCategoryIdsSplited = orderCategoryIds.split(",");

                    List<Tuple2<Long, Long>> list = new ArrayList<Tuple2<Long, Long>>();
                    for (String orderCategoryId : orderCategoryIdsSplited) {
                        list.add(new Tuple2<Long, Long>(Long.valueOf(orderCategoryId), 1L));
                    }
                    return list;
                }
            });

        JavaPairRDD<Long, Long> orderCategoryIdCount = orderCategoryIdRDD.reduceByKey(
            new Function2<Long, Long, Long>() {
                @Override
                public Long call(Long v1, Long v2) throws Exception {
                    return v1 + v2;
                }
            });

        return orderCategoryIdCount;
    }

    private static JavaPairRDD<Long, Long> getPayCategoryId2CountRDD(JavaPairRDD<String, Row> sessionid2detailRDD) {
        JavaPairRDD<String, Row> payActionRDD = sessionid2detailRDD.filter(
            new Function<Tuple2<String, Row>, Boolean>() {
                private static final long serialVersionUID = 369423122908077599L;

                @Override
                public Boolean call(Tuple2<String, Row> tuple) throws Exception {
                    Row row = tuple._2;
                    return row.getString(10) != null ? true : false;
                }
            });

        JavaPairRDD<Long, Long> payCategoryIdRDD = payActionRDD.flatMapToPair(
            new PairFlatMapFunction<Tuple2<String, Row>, Long, Long>() {
                private static final long serialVersionUID = -6636902634726016454L;

                @Override
                public Iterable<Tuple2<Long, Long>> call(Tuple2<String, Row> tuple) throws Exception {
                    Row row = tuple._2;
                    String payCategoryIds = row.getString(10);
                    String[] payCategoryIdsSplited = payCategoryIds.split(",");

                    List<Tuple2<Long, Long>> list = new ArrayList<Tuple2<Long, Long>>();
                    for(String payCategoryId : payCategoryIdsSplited){
                        list.add(new Tuple2<Long, Long>(Long.valueOf(payCategoryId), 1L));
                    }
                    return list;
                }
            });

        JavaPairRDD<Long, Long> payCategoryIdCount = payCategoryIdRDD.reduceByKey(
            new Function2<Long, Long, Long>() {
                @Override
                public Long call(Long v1, Long v2) throws Exception {
                    return v1 + v2;
                }
            });

        return payCategoryIdCount;
    }

    private static JavaPairRDD<Long, String> joinCategoryAndData(
            JavaPairRDD<Long, Long> categoryId2CountRDD,
            JavaPairRDD<Long, Long> clickCategoryId2CountRDD,
            JavaPairRDD<Long, Long> orderCategoryId2CountRDD,
            JavaPairRDD<Long, Long> payCategoryId2CountRDD){

        JavaPairRDD<Long, Tuple2<Long, Optional<Long>>> tmpJoinRDD =
                categoryId2CountRDD.leftOuterJoin(clickCategoryId2CountRDD);

        JavaPairRDD<Long, String> tmpMapRDD = tmpJoinRDD.mapToPair(
            new PairFunction<Tuple2<Long, Tuple2<Long, Optional<Long>>>, Long, String>() {
                private static final long serialVersionUID = 4503608897308291618L;

                @Override
                public Tuple2<Long, String> call(
                        Tuple2<Long, Tuple2<Long, Optional<Long>>> tuple) throws Exception {
                    long categoryId = tuple._1;
                    Optional<Long> optional = tuple._2._2;
                    long clickCount = 0L;
                    if(optional.isPresent()){
                        clickCount = optional.get();
                    }
                    String value = Constants.FIELD_CATEGORY_ID + "=" + categoryId + "|" +
                            Constants.FIELD_CLICK_COUNT + "=" + clickCount;

                    return new Tuple2<Long, String>(categoryId, value);
                }
        });

        tmpMapRDD = tmpMapRDD.leftOuterJoin(orderCategoryId2CountRDD).mapToPair(
            new PairFunction<Tuple2<Long, Tuple2<String, Optional<Long>>>, Long, String>() {
                private static final long serialVersionUID = -9085025528722020242L;

                @Override
                public Tuple2<Long, String> call(
                        Tuple2<Long, Tuple2<String, Optional<Long>>> tuple) throws Exception {
                    long categoryId = tuple._1;
                    String value = tuple._2._1;
                    Optional<Long> optional = tuple._2._2;
                    long orderCount = 0L;
                    if(optional.isPresent()){
                        orderCount = optional.get();
                    }
                    value = value + "|" +
                            Constants.FIELD_ORDER_COUNT + "=" + orderCount;

                    return new Tuple2<Long, String>(categoryId, value);
                }
        });

        tmpMapRDD = tmpMapRDD.leftOuterJoin(payCategoryId2CountRDD).mapToPair(
                new PairFunction<Tuple2<Long, Tuple2<String, Optional<Long>>>, Long, String>() {
                    private static final long serialVersionUID = 1821372949480798669L;

                    @Override
                    public Tuple2<Long, String> call(
                            Tuple2<Long, Tuple2<String, Optional<Long>>> tuple) throws Exception {
                        long categoryId = tuple._1;
                        String value = tuple._2._1;
                        Optional<Long> optional = tuple._2._2;
                        long payCount = 0L;
                        if(optional.isPresent()){
                            payCount = optional.get();
                        }
                        value = value + "|" +
                                Constants.FIELD_PAY_COUNT + "=" + payCount;

                        return new Tuple2<Long, String>(categoryId, value);
                    }
                });

        return tmpMapRDD;
    }

    private static void getTop10Session(JavaSparkContext sc, List<Tuple2<CategorySortKey, String>> top10CategoryList,
                                        JavaPairRDD<String, Row> sessionid2detailRDD, final long taskId) {

        /**
         * Top10热门品类RDD
         */
        List<Tuple2<Long, Long>> top10CategoryIdList = new ArrayList<Tuple2<Long, Long>>();
        for(Tuple2<CategorySortKey, String> category : top10CategoryList){
            long categoryid = Long.valueOf(StringUtils.getFieldFromConcatString(category._2,
                    "\\|", Constants.FIELD_CATEGORY_ID));
            top10CategoryIdList.add(new Tuple2<Long, Long>(categoryid, categoryid));
        }

        JavaPairRDD<Long, Long> top10CategoryIdRDD = sc.parallelizePairs(top10CategoryIdList);

        /**
         * Top10品类对被各session点击的次数 <categoryid, session:count>
         */
        JavaPairRDD<String, Iterable<Row>> sessionid2detailsRDD= sessionid2detailRDD.groupByKey();
        JavaPairRDD<Long, String> categoryid2sessionCountRDD = sessionid2detailsRDD.flatMapToPair(
            new PairFlatMapFunction<Tuple2<String, Iterable<Row>>, Long, String>() {
                private static final long serialVersionUID = -3714231740357622693L;

                @Override
                public Iterable<Tuple2<Long, String>> call(Tuple2<String, Iterable<Row>> tuple) throws Exception {
                    String sessionid = tuple._1;
                    Iterator<Row> iterator = tuple._2.iterator();

                    Map<Long, Long> categoryCountMap = new HashMap<Long, Long>();

                    // 计算每个品类的点击次数
                    while(iterator.hasNext()){
                        Row row = iterator.next();
                        if(row.get(6) != null){
                            long categoryid = row.getLong(6);

                            Long count = categoryCountMap.get(categoryid);
                            if(count == null){
                                count = 0L;
                            }
                            count ++;
                            categoryCountMap.put(categoryid, count);
                        }
                    }

                    List<Tuple2<Long, String>> list = new ArrayList<Tuple2<Long, String>>();
                    for(Map.Entry<Long, Long> categoryCountEntry : categoryCountMap.entrySet()){
                        long categoryid = categoryCountEntry.getKey();
                        long count = categoryCountEntry.getValue();
                        String value = sessionid + "," + count;
                        list.add(new Tuple2<Long, String>(categoryid, value));
                    }

                    return list;
                }
            }
        );

        // 热门品类被各个session点击的次数
        JavaPairRDD<Long, String> top10CategorySessionCountRDD = top10CategoryIdRDD
            .join(categoryid2sessionCountRDD)
            .mapToPair(new PairFunction<Tuple2<Long, Tuple2<Long, String>>, Long, String>() {
                private static final long serialVersionUID = 8986238215832529287L;

                @Override
                public Tuple2<Long, String> call(Tuple2<Long, Tuple2<Long, String>> tuple) throws Exception {
                    return new Tuple2<Long, String>(tuple._1, tuple._2._2);
                }
            });

        /**
         * 分组取Top10,获取Top10的活跃用户
         */
        JavaPairRDD<Long, Iterable<String>> top10CategorySessionCountsRDD =
                top10CategorySessionCountRDD.groupByKey();
        JavaPairRDD<String, String> top10SessionRDD = top10CategorySessionCountsRDD.flatMapToPair(
            new PairFlatMapFunction<Tuple2<Long, Iterable<String>>, String, String>() {
                private static final long serialVersionUID = -7813217754576330603L;

                @Override
                public Iterable<Tuple2<String, String>> call(
                        Tuple2<Long, Iterable<String>> tuple) throws Exception {
                    long categoryid = tuple._1;
                    Iterator<String> iterator = tuple._2.iterator();
                    String[] top10Sessions = new String[10];

                    while(iterator.hasNext()){
                        String sessionCount = iterator.next();
                        long count = Long.valueOf(sessionCount.split(",")[1]);

                        for(int i = 0;i < top10Sessions.length; i ++){
                            if(top10Sessions[i] == null){
                                top10Sessions[i] = sessionCount;
                                break;
                            } else {
                                long _count = Long.valueOf(top10Sessions[i].split(",")[1]);
                                if(count > _count){
                                    for(int j = 9; j > i; j --){
                                        top10Sessions[j] = top10Sessions[j - 1];
                                    }
                                    top10Sessions[i] = sessionCount;
                                    break;
                                }
                            }
                        }
                    }

                    List<Tuple2<String, String>> list = new ArrayList<Tuple2<String, String>>();
                    ITop10SessionDao top10SessionDao = DAOFactory.getTop10SessionDao();

                    for(String sessionCount : top10Sessions){
                        if(sessionCount != null){
                            String sessionid = sessionCount.split(",")[0];
                            long count = Long.valueOf(sessionCount.split(",")[1]);
                            Top10Session top10Session = new Top10Session();
                            top10Session.setTaskid(taskId);
                            top10Session.setCategoryid(categoryid);
                            top10Session.setSessionid(sessionid);
                            top10Session.setClickCount(count);
                            top10SessionDao.insert(top10Session);
                            list.add(new Tuple2<String, String>(sessionid, sessionid));
                        }
                    }
                    return list;
                }
        });

        /**
         * 获取top10活跃session的明细数据
         */
        JavaPairRDD<String, Tuple2<String, Row>> sessionDetailRDD =
                top10SessionRDD.join(sessionid2detailRDD);
        sessionDetailRDD.foreach(new VoidFunction<Tuple2<String,Tuple2<String,Row>>>() {

            private static final long serialVersionUID = 1L;

            @Override
            public void call(Tuple2<String, Tuple2<String, Row>> tuple) throws Exception {
                Row row = tuple._2._2;

                SessionDetail sessionDetail = new SessionDetail();
                sessionDetail.setTaskid(taskId);
                sessionDetail.setUserid(row.getLong(1));
                sessionDetail.setSessionid(row.getString(2));
                sessionDetail.setPageid(row.getLong(3));
                sessionDetail.setActionTime(row.getString(4));
                sessionDetail.setSearchKeyWord(row.getString(5));
                sessionDetail.setClickCategoryId(row.getLong(6));
                sessionDetail.setClickProductId(row.getLong(7));
                sessionDetail.setOrderCategoryIds(row.getString(8));
                sessionDetail.setOrderProductIds(row.getString(9));
                sessionDetail.setPayCategoryIds(row.getString(10));
                sessionDetail.setPayProductIds(row.getString(11));

                ISessionDetailDao sessionDetailDAO = DAOFactory.getSessionDetailDao();
                sessionDetailDAO.insert(sessionDetail);
            }
        });
    }

}
